package com.atguigu.day11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author Felix
 * @date 2024/7/22
 * 该案例演示了FlinkSQL整体开发流程
 */
public class Flink01_Env {
    public static void main(String[] args) {
        //TODO 1. 环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(1);
        //1.3 指定表执行环境
        //StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, EnvironmentSettings.newInstance().build());
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //TODO 2.~~~~~~~流--->动态表~~~~~~~~
        //DataStreamSource<Integer> ds = env.fromElements(1, 2, 3, 4);

        // 2.1 方式1  通过TableApi
        //Table numTable = tableEnv.fromDataStream(ds, $("num"));

        //2.2 方式2  直接执行SQL
        tableEnv.executeSql("CREATE TABLE source ( \n" +
                "    id INT, \n" +
                "    ts BIGINT, \n" +
                "    vc INT \n" +
                ") WITH ( \n" +
                "    'connector' = 'datagen', \n" +
                "    'rows-per-second'='1', \n" +
                "    'fields.id.kind'='random', \n" +
                "    'fields.id.min'='1', \n" +
                "    'fields.id.max'='10', \n" +
                "    'fields.ts.kind'='sequence', \n" +
                "    'fields.ts.start'='1', \n" +
                "    'fields.ts.end'='1000000', \n" +
                "    'fields.vc.kind'='random', \n" +
                "    'fields.vc.min'='1', \n" +
                "    'fields.vc.max'='100'\n" +
                ");\n");
        //TODO 3. ~~~~~~~持续查询--->动态表~~~~~~~~~~
        //3.1 通过SQL的方式
        //如果直接查询，会报错，表对象没有找到
        //tableEnv.executeSql("select * from numTable").print();
        //将表对象注册到表执行环境中
        //tableEnv.registerTable("num_table",numTable);
        //tableEnv.createTemporaryView("num_table",numTable);
        //tableEnv.executeSql("select * from num_table").print();
        //tableEnv.executeSql("select * from " + numTable).print();

        //tableEnv.sqlQuery("select * from num_table").execute().print();
        //TableResult tableResult = tableEnv.executeSql("select * from num_table");

        //tableEnv.sqlQuery("select * from source").execute().print();

        //注意：如果读取的是无界数据，FlinkSQL的print方法是不会结束的，所以在程序中一般只会保留一个print
        //tableEnv.sqlQuery("select * from xxxx").execute().print();

        //3.1 通过API的方式
        Table sourceTable = tableEnv.sqlQuery("select * from source");
        sourceTable
                .select($("id"),$("ts"),$("vc"))
                .where($("id").isEqual(1)).execute().print();



        //+----+-------------+
        //| op |         num |
        //+----+-------------+
        //| +I |           1 |
        //| +I |           2 |
        //| +I |           3 |
        //| +I |           4 |
        //+----+-------------+
        //        4 rows in set

        //3.2 通过API的方式


        //TODO 提交作业
        //如果使用的是DataStreamAPI，最后需要通过env.execute方法提交作业
        //env.execute();
        //注意：如果最后操作的是动态表，不需要显示的提交作业
    }
}
